深入理解 Gem5 之一

近期研究需要，我开始研究 gem5 模拟器的底层实现 。gem5 模拟器是一款模块化的计算机系统架构研究平台，可用于研究系统级架构、处理器微架构。gem5 是一个具有开放治理模型的社区主导项目，最初是为学术界的计算机体系结构研究而构想的，目前已发展为学术界、工业研究和教学中的计算机系统设计。

根据 gem5 的 paper，gem5 框架融合了 M5 和 GEMS 两者的实现。其中 M5 提供高度可配置的仿真框架，包含了多个 ISA 和多种 CPU 模型；而 GEMS 的详细而灵活的内存系统提供了对多个缓存一致性协议和互连模型的支持。目前，gem5 支持大多数商业 ISA（ARM、ALPHA、MIPS、Power、SPARC 和 x86），包括在其中三个（ARM、ALPHA 和 x86）上 booting Linux。

该项目是许多学术和工业机构共同努力的结果，包括 AMD、ARM、HP、MIPS、普林斯顿大学、麻省理工学院以及密歇根大学、德克萨斯大学和威斯康星大学。在过去的十年中，M5 和 GEMS 已在数百种出版物中使用，并已被下载数万次。 gem5 项目上的高水平协作，再加上组件部分的先前成功和类似 BSD 的自由许可证，使 gem5 成为一个有价值的全系统仿真工具。

在本博客中，我将探讨如何创建、调度事件，并深入理解背后的原理。

gem5 是一个事件驱动（Event-driven）的模拟器。在事件驱动模型中，每个事件（Event）都有一个回调函数用于处理事件。

下面以 HelloObject 类代码为例，在代码中添加一个事件触发时执行的新函数 processEvent()。此函数必须不带参数并且不返回任何内容。然后，添加一个 EventFunctionWrapper 类对象 event，并在构造函数中包装 processEvent()。最后，添加了一个 startup() 函数，其中使用 schedule() 函数开始调度事件，即让该事件安排在未来的某个时刻被触发（事件驱动的模拟不允许事件在过去执行）。

HelloObject::HelloObject(HelloObjectParams *params) : SimObject(params), event([this]{processEvent();}, name()) { DPRINTF(Hello, "Created the hello object\n");}

void HelloObject::startup() { schedule(event, 100);}

void HelloObject::processEvent() { DPRINTF(Hello, "Hello world! Processing the event!\n");}

上面的代码中，事件会在第 100 个 tick 时被触发。通常，需使用 curTick() 加时间偏移来确定事件触发的时间。但在这一简单示例中，开始模拟的函数（即 Python 配置文件中调用 simulate() 函数） 是以 tick = 0 的原点开始执行的，因此这里的 startup() 可以显式地标明要调度的时间点。

当运行 gem5 模拟器后，会得到如下输出，具体如何运行不是本博客的重点，感兴趣的读者可参考官方文档运行。从下面的输出信息可知，我们实现了在 100 tick 时执行 processEvent() 函数。

gem5 compiled Jan 4 2017 11:01:46gem5 started Jan 4 2017 13:41:38gem5 executing on chinook, pid 1834command line: build/X86/gem5.opt --debug-flags=Hello configs/learning_gem5/part2/run_hello.py

Global frequency set at 1000000000000 ticks per second 0: hello: Created the hello objectBeginning simulation!info: Entering event queue @ 0. Starting simulation... 100: hello: Hello world! Processing the event!Exiting @ tick 18446744073709551615 because simulate() limit reached

当然，processEvent() 也可以很复杂，甚至在执行 processEvent() 时再加入一个新的事件等待被触发：

class EventFunctionWrapper : public Event { private: std::function callback; std::string _name; public: EventFunctionWrapper(const std::function &callback, const std::string &name, bool del = false, Priority p = Default_Pri) : Event(p), callback(callback), _name(name) { if (del) setFlags(AutoDelete); } void process() { callback(); }};

事件队列（EventQueue）是管理系统事件的重要载体，类 EventQueue 用于描述该事件队列，它将被调度的事件按执行时间排列，其中队列头是执行时间最早的事件，执行时间一致的按照优先级高低在另一个维度上排列，形成了一个二维链表。每个线程都会维护一个局部的事件队列，事件会在队列中被调度（即插入到队列中）。

async_queue 和 async_queue_mutex 用于管理多线程下异步事件队列，而真正的链表头是 head。

EventQueue 类重点还是在于 schedule() 函数，该函数负责事件调度，将创建好的事件插入到事件队列中，以备事件引擎执行。其参数是将要被执行的事件 event 和具体执行时间 when，global 参数用于判断队列是否被另一个线程运行。在 schedule() 中，设置完 event 的执行时间后，分成两个执行方式：同步插入和异步插入。

void schedule(Event *event, Tick when, bool global=false) { event->setWhen(when, this); // 两种模式: // a. 异步插入：硬件线程将局部事件调度到一个不属于自己的其他队列 需要 `asyncq`. // b. 全局调度：硬件线程将全局事件调度到 `asyncq` 需要维护全局事件的整体顺序 // See global_event.{cc,hh} for more explanation. if (inParallelMode && (this != curEventQueue() || global)) { // 异步插入 asyncInsert(event); } else { // 同步插入 insert(event); } event->flags.set(Event::Scheduled); event->acquire();

if (debug::Event) event->trace("scheduled");}

同步事件的调度需要调用 insert() 函数，且此时 global 参数为 false。这是很简单的链表插入动作，插入在插入点之前，符合编程惯例。之前提到，EventQueue 队列是一个二级链表，是由一串串单链表的头节点组成的链表。从以下代码可知，在 in Bin 链表中，使用头插法插入了新节点，因此二级链表是 LIFO 的。

void EventQueue::insert(Event *event) { if (!head || *event nextBin; while (curr && *curr < *event) { prev = curr; curr = curr->nextBin; } // Note: this operation may render all nextBin pointers on the // prev 'in bin' list stale (except for the top one) prev->nextBin = Event::insertBefore(event, curr);}

跨线程的事件调度可采用异步事件的调度方案（即将事件调度到另一个线程的事件队列中），此时 global 参数设置为 true。采取该做法时，因为事件队列被另一个线程运行，为防止冲突，asyncInsert() 会将事件暂时插入到异步队列 async_queue 中。

void EventQueue::handleAsyncInsertions() { assert(this == curEventQueue()); async_queue_mutex.lock(); while (!async_queue.empty()) { insert(async_queue.front()); async_queue.pop_front(); } async_queue_mutex.unlock();}

最后在每个模拟循环（或barrier）的最后时刻，async_queue 中的事件会被全部取出，然后捯饬到真正的事件队列中（见上述 handleAsyncInsertions() 函数）。

若要直接跨线程和事件队列地调度事件，或者获取目标事件队列的锁，在 schedule() 之外，还需要有特殊的机制避免死锁。原理也很简单：线程必须先自动放弃当前手中的事件队列锁，才能获得新事件队列锁，确保每个线程至多获取一个事件队列锁。该功能由 ScopedMigration 类负责，它将事件从一个事件队列暂时移植到另一个队列中。

在 schedule() 函数的基础之上，还添加了 reschedule() 和 deschedule() 等函数，方便事件队列的调度管理。其中 deschedule() 函数牵扯到链表的删除，也是比较平凡的实现。

Event *Event::removeItem(Event *event, Event *top) { Event *curr = top; Event *next = top->nextInBin; // if we removed the top item, we need to handle things specially // and just remove the top item, fixing up the next bin pointer of // the new top item if (event == top) { if (!next) return top->nextBin; next->nextBin = top->nextBin; return next; } // Since we already checked the current element, we're going to // keep checking event against the next element. while (event != next) { if (!next) panic("event not found!");

curr = next; next = next->nextInBin; } // remove next from the 'in bin' list since it's what we're looking for curr->nextInBin = next->nextInBin; return top;}

现在我们来看一下 EventQueue 中，Event 是如何被执行的：

if (next) { next->nextBin = head->nextBin; head = next; } else head = head->nextBin; if (!event->squashed()) { // forward current cycle to the time when this event occurs. setCurTick(event->when()); if (debug::Event) event->trace("executed"); event->process(); if (event->isExitEvent()) { assert(!event->flags.isSet(Event::Managed) || !event->flags.isSet(Event::IsMainQueue)); // would be silly return event; } } else event->flags.clear(Event::Squashed); event->release(); return NULL;}

当调用 ServiceOne() 函数时，默认是处理 EventQueue 中的第一个事件，取出第一个事件 event 后，在判断其是否 squashed（被压缩、合并），若不是被压缩的，那么直接将时间调整至事件处理时刻（事件驱动），随后调用 event->process() ，若是，那么仅需要清除标志位即可。

为降低用户使用难度，使用 EventManager 类来包装 EventQueue 类，包括常用的许多事件调度函数，而这些包装函数是 SimObject 类内经常调用的：

void deschedule(Event &event) { eventq->deschedule(&event); }};

现在，再回到 HelloObject 类上来，HelloObject 类继承了 SimObject 类。值得注意的是，SimObject 类是 gem5 中最重要的类之一，几乎所有的系统组件，如 CPU 、总线、Cache 等都需要继承 SimObject 类。根据代码，SimObject 类继承了 EventManager 类，EventManager 类中包装了一个 EventQueue 对象。因此，在 startup() 函数中使用的 schedule() 函数本质就是调用了 EventQueue::schedule() 函数：将事件 event 放入事件队列中，等待被调度。

另外，EventFunctionWrapper 类是 Event 类的子类，该类建立起了回调函数与事件对象的联系，当事件被触发后，回调函数会被执行。

HelloObject::HelloObject(HelloObjectParams *params) : SimObject(params), event([this]{processEvent();}, name()) { DPRINTF(Hello, "Created the hello object\n");}

void HelloObject::startup() { schedule(event, 100);}